package com.rrd.dw.mr.gzip;

import java.io.IOException;
import java.io.InputStream;
import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.google.gson.Gson;
import com.rrd.dw.utils.HBaseUtil;

/**
 * Created by liuchuanqi on 2018/01/24.
 */
public class ExportTaobaoDataToDW extends Configured implements Tool {

    public static final String[] TAOBAO_COLUMNS = { "nick_name", "real_name",
        "birthday", "sex", "login_name", "source_user_mobile",
        "source_user_idno", "source" };//taobao
    public static final String[] ORDER_COLUMNS = { "amount", "time:date",
        "order_no", "order_quantity", "freight", "order_status", "address" };//taobao

    public static final String[] ADDRESS_COLUMNS = { "consignee", "province",
        "city", "full_address", "post_code", "mobile" };//taobao

    public static final String[] PRODUCT_COLUMNS = { "shop_name", "original",
        "promotion", "name", "quantity:int", "attr", "service_info" };//taobao
    public static final String[] PAY_COLUMNS = { "mess_type", "red_pack",
        "zhifubao_amount", "bankcard_count", "huabei_limit", "jifenbao" };//pay

    public static final String[] JOB_COLUMNS = { "finished:boolean",
        "requested:boolean", "successful:boolean", "start_time:date",
        "end_time:date", "system_id", "query", "try_times:int", "type",
        "message", "code" };//job

    @Override
    public int run(String[] args) throws Exception {
        Properties prop = new Properties();
        try (InputStream is = this.getClass().getClassLoader()
            .getResourceAsStream("hbase.properties");) {
            prop.load(is);
        }

        Job job = Job.getInstance(HBaseConfiguration.create(this.getConf()));
        Configuration conf = job.getConfiguration();
        conf.set("hbase.zookeeper.quorum",
            prop.get("hbase.zookeeper.quorum").toString());
        conf.set("hbase.rootdir", prop.get("hbase.rootdir").toString());

        conf.set("mapreduce.map.memory.mb", "3072");
        conf.set("mapreduce.reduce.memory.mb", "5120");
        conf.set("hbase.client.scanner.timeout.period", "120");
        conf.set("mapreduce.output.fileoutputformat.compress", "true");
        conf.set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");
        conf.set("mapreduce.output.fileoutputformat.compress.codec",
            "org.apache.hadoop.io.compress.GzipCodec");
        conf.set("io.compression.codecs",
            "org.apache.hadoop.io.compress.GzipCodec");
        conf.set("mapreduce.task.io.sort.factor", "10");
        String[] otherArgs = new GenericOptionsParser(conf, args)
            .getRemainingArgs();
        if (otherArgs.length != 3) {
            System.err
                .println("Usage: DataExportTool <out> <start_time> <end_time>");
            return -1;
        }
        Long startTime = HBaseUtil.parseDate(otherArgs[1]).getTime();
        Long endTime = HBaseUtil.parseDate(otherArgs[2]).getTime();
        job.setJarByClass(ExportTaobaoDataToDW.class);
        List<Scan> scans = new ArrayList<>();

        Scan scan = new Scan();
        scan.setCacheBlocks(false);
        scan.setCaching(200);
        scan.setMaxVersions(1);
        scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME,
            Bytes.toBytes("alitaobao_records"));
        scan.setTimeRange(startTime, endTime + 14400000);
        scans.add(scan);

        TableMapReduceUtil.initTableMapperJob(scans,
            ExportTaobaoDataToDWMapper.class, Text.class, Text.class, job);
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[0]));
        MultipleOutputs.addNamedOutput(job, "basepayinfo",
            TextOutputFormat.class, Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job, "orderlist", TextOutputFormat.class,
            Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job, "prodlist", TextOutputFormat.class,
            Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job, "addresslist",
            TextOutputFormat.class, Text.class, Text.class);

        job.setReducerClass(ExportTaobaoDataToDWReducer.class);
        job.setNumReduceTasks(5);
        return job.waitForCompletion(true) ? 0 : 1;

    }

    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        int res = ToolRunner.run(conf, new ExportTaobaoDataToDW(), args);
        System.exit(res);
    }

    public static class ExportTaobaoDataToDWMapper
            extends TableMapper<Text, Text> {

        @Override
        protected void map(ImmutableBytesWritable key, Result value,
                Context context) throws IOException, InterruptedException {

            StringBuffer sb = new StringBuffer("");
            Object[] rowkey = HBaseUtil.decodeRowkey(key.get());
            if (rowkey == null) {
                return;
            }
            long timestamp = Long.parseLong(rowkey[0].toString());
            String jobid = rowkey[1].toString();
            Cell dataCell = null;
            Cell countCell = null;
            String type = "";
            String field = "";

            //淘宝基本信息
            for (String str1 : TAOBAO_COLUMNS) {
                String[] fieldArray = str1.split(":");
                if (fieldArray.length > 1) {
                    field = fieldArray[0];
                    type = fieldArray[1];
                } else {
                    field = fieldArray[0];
                    type = "string";
                }
                dataCell = value.getColumnLatestCell(Bytes.toBytes("taobao"),
                    Bytes.toBytes(field));
                if (dataCell != null) {
                    sb.append(HBaseUtil.getValue(type,
                        CellUtil.cloneValue(dataCell)));
                }
                sb.append("\t");
            }
            //pay信息
            for (String str1 : PAY_COLUMNS) {
                String[] fieldArray = str1.split(":");
                if (fieldArray.length > 1) {
                    field = fieldArray[0];
                    type = fieldArray[1];
                } else {
                    field = fieldArray[0];
                    type = "string";
                }
                dataCell = value.getColumnLatestCell(Bytes.toBytes("pay"),
                    Bytes.toBytes(field));
                if (dataCell != null) {
                    sb.append(HBaseUtil.getValue(type,
                        CellUtil.cloneValue(dataCell)));
                }
                sb.append("\t");
            }
            //job信息
            for (String str1 : JOB_COLUMNS) {
                String[] fieldArray = str1.split(":");
                if (fieldArray.length > 1) {
                    field = fieldArray[0];
                    type = fieldArray[1];
                } else {
                    field = fieldArray[0];
                    type = "string";
                }
                dataCell = value.getColumnLatestCell(Bytes.toBytes("job"),
                    Bytes.toBytes(field));
                if (dataCell != null) {
                    sb.append(HBaseUtil.getValue(type,
                        CellUtil.cloneValue(dataCell)));
                }
                sb.append("\t");
            }
            String outString = sb.toString();
            context.write(new Text(timestamp + "\t" + jobid),
                new Text(outString.substring(0, outString.lastIndexOf("\t"))));
            sb = null;
            //订单信息
            countCell = value.getColumnLatestCell(Bytes.toBytes("taobao"),
                Bytes.toBytes("AliTaobaoOrdercount"));
            if (countCell != null) {
                int count = Bytes.toInt(CellUtil.cloneValue(countCell));
                for (int i = 0; i < count; i++) {
                    String orderNo = null;
                    sb = new StringBuffer("");
                    for (String bstr : ORDER_COLUMNS) {
                        String[] fieldArray = bstr.split(":");
                        if (fieldArray.length > 1) {
                            field = fieldArray[0];
                            type = fieldArray[1];
                        } else {
                            field = fieldArray[0];
                            type = "string";
                        }
                        dataCell = value.getColumnLatestCell(
                            Bytes.toBytes("taobao"),
                            Bytes.toBytes(field + "#" + i));
                        if (dataCell != null) {
                            sb.append(HBaseUtil.getValue(type,
                                CellUtil.cloneValue(dataCell)));
                            if ("order_no".equals(bstr)) {
                                orderNo = HBaseUtil
                                    .getValue(type,
                                        CellUtil.cloneValue(dataCell))
                                    .toString();
                            }
                        }
                        sb.append("\t");
                    }
                    outString = sb.toString();
                    context.write(new Text(timestamp + "\t" + jobid), new Text(
                        outString.substring(0, outString.lastIndexOf("\t"))));
                    sb = null;

                    //商品
                    countCell = value.getColumnLatestCell(
                        Bytes.toBytes("taobao"),
                        Bytes.toBytes("AliTaobaoOrderProductcount#" + i));
                    if (countCell != null) {
                        int pcount = Bytes
                            .toInt(CellUtil.cloneValue(countCell));
                        for (int j = 0; j < pcount; j++) {
                            sb = new StringBuffer("");
                            for (String bstr : PRODUCT_COLUMNS) {
                                String[] fieldArray = bstr.split(":");
                                if (fieldArray.length > 1) {
                                    field = fieldArray[0];
                                    type = fieldArray[1];
                                } else {
                                    field = fieldArray[0];
                                    type = "string";
                                }
                                dataCell = value.getColumnLatestCell(
                                    Bytes.toBytes("taobao"),
                                    Bytes.toBytes(field + "#" + i + "#" + j));
                                if (dataCell != null) {
                                    sb.append(HBaseUtil.getValue(type,
                                        CellUtil.cloneValue(dataCell)));
                                }
                                sb.append("\t");
                            }
                            outString = sb.toString();
                            context.write(
                                new Text(
                                    timestamp + "\t" + jobid + "\t" + orderNo),
                                new Text(outString.substring(0,
                                    outString.lastIndexOf("\t"))));
                            sb = null;
                        }
                    }
                }
            }
            //地址信息
            countCell = value.getColumnLatestCell(Bytes.toBytes("taobao"),
                Bytes.toBytes("AliTaobaoAddresscount"));
            if (countCell != null) {
                int count = Bytes.toInt(CellUtil.cloneValue(countCell));
                for (int i = 0; i < count; i++) {
                    sb = new StringBuffer("");
                    for (String bstr : ADDRESS_COLUMNS) {
                        String[] fieldArray = bstr.split(":");
                        if (fieldArray.length > 1) {
                            field = fieldArray[0];
                            type = fieldArray[1];
                        } else {
                            field = fieldArray[0];
                            type = "string";
                        }
                        dataCell = value.getColumnLatestCell(
                            Bytes.toBytes("taobao"),
                            Bytes.toBytes(field + "#" + i));
                        if (dataCell != null) {
                            sb.append(HBaseUtil.getValue(type,
                                CellUtil.cloneValue(dataCell)));
                        }
                        sb.append("\t");
                    }
                    outString = sb.toString();
                    context.write(new Text(timestamp + "\t" + jobid), new Text(
                        outString.substring(0, outString.lastIndexOf("\t"))));
                    sb = null;
                }
            }
        }

    }

    public static class ExportTaobaoDataToDWReducer
            extends Reducer<Text, Text, Text, Text> {
        private MultipleOutputs<Text, Text> mos;

        @Override
        protected void setup(Context context)
                throws IOException, InterruptedException {
            this.mos = new MultipleOutputs<>(context);
        }

        @Override
        protected void cleanup(Context context)
                throws IOException, InterruptedException {
            this.mos.close();
        }

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            for (Text val : values) {
                int ll = val.toString().split("\t", -1).length;
                if (ll > 10) {
                    this.mos.write("basepayinfo", key, val);
                } else if (key.toString().split("\t").length > 2) {
                    this.mos.write("prodlist", key, val);
                } else if (ll == 6) {
                    this.mos.write("addresslist", key, val);
                } else {
                    this.mos.write("orderlist", key, val);
                }
            }

        }
    }
}
